package com.abcdefg.demo.rpc;

import com.abcdefg.demo.constants.Constants;
import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeoutException;

public class RpcClient {
    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        //1.建立连接
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(Constants.Host);
        factory.setPort(Constants.Port);
        factory.setVirtualHost(Constants.VirtualHost);
        factory.setUsername(Constants.UserName);
        factory.setPassword(Constants.Password);
        Connection connection = factory.newConnection();
        //2.创建信道
        Channel channel = connection.createChannel();
        channel.queueDeclare(Constants.RPC_REQUEST_QUEUE2,true,false,false,null);
        channel.queueDeclare(Constants.RPC_RESPONSE_QUEUE1,true,false,false,null);


        //唯⼀标志本次请求
        String corrId = UUID.randomUUID().toString();
        AMQP.BasicProperties props = new AMQP.BasicProperties
                .Builder()
                .correlationId(corrId)
                .replyTo(Constants.RPC_RESPONSE_QUEUE1)
                .build();
        String msg="hello rpc...";
        channel.basicPublish("",Constants.RPC_REQUEST_QUEUE2,props,msg.getBytes());

        //处理响应
        //定义阻塞队列，用来接受响应
        final BlockingQueue<String> response = new ArrayBlockingQueue<>(1);
        DefaultConsumer consumer=new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                if(corrId.equals(properties.getCorrelationId())){
                    response.add(new String(body));
                }
            }
        };
        channel.basicConsume(Constants.RPC_RESPONSE_QUEUE1,true,consumer);
        String result = response.take();
        System.out.println("响应的结果："+result);

        connection.close();
    }
}
